package com.mininglamp.content

import com.mongodb.spark.MongoSpark
import com.mongodb.spark.config.ReadConfig
import org.apache.spark.SparkConf
import org.apache.spark.ml.feature.{HashingTF, IDF, Tokenizer}
import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.jblas.DoubleMatrix

/**
  * Project: ECommerceRecommendSystem
  * Package: com.mininglamp.content
  * Description:基于内容的相似推荐
  *
  * Created by ZhouPeng on 2022/01/11 10:30
  **/
object ContentRecommender {

  //定义mongo中product集合
  val MONGODB_PRODUCT_COLLECTION = "Product"
  //定义基于内容推荐的相似商品列表集合
  val CONTENT_PRODUCT_RECS_COLLECTION = "ContentBasedProductsRecs"

  def main(args: Array[String]): Unit = {

    //基本配置
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://master:27017/recommender",
      "mongo.db" -> "recommender"
    )

    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("ContentRecommender")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    import sparkSession.implicits._
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    val readConfig = ReadConfig(Map(
      "uri" -> mongoConfig.mongoUri,
      "collection" -> MONGODB_PRODUCT_COLLECTION
    ))

    //加载products数据
    val productTagsDF = MongoSpark.load(sparkSession.sparkContext, readConfig)
      .map(
        document => {
          (document.getInteger("productId"), document.getString("name"),
            document.getString("tags").map(x => if (x == '|' || x == "\\") ' ' else x))
        }
      ).toDF("productId", "name", "tags").cache()

    //todo 核心计算过程。使用 TF-IDF 算法提取商品特征向量
    //1. 实例化一个分词器，用来分词，默认空格做分割
    val tokenizer = new Tokenizer().setInputCol("tags").setOutputCol("words")
    val wordsDataDF = tokenizer.transform(productTagsDF)

    //2.定义HashingTF 工具，计算频次
    val hashingTF = new HashingTF().setInputCol("words").setOutputCol("rawFeatures")
      //设置特征向量维度，也就是hash分桶个数
      .setNumFeatures(800)
    val featurizeDataDF = hashingTF.transform(wordsDataDF)
    //featurizeDataDF.show(truncate = false)

    //3. 定义一个IDF工具，计算 TF-IDF
    val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
    //训练 idf 模型
    val iDFModel = idf.fit(featurizeDataDF)
    //得到新增加列features的DF
    val rescaledDataDF = iDFModel.transform(featurizeDataDF)
    //rescaledDataDF.show(truncate = false)

    //对DF进行转换，得到rdd形式的features
    val productFeatures = rescaledDataDF.map(
      row => {
        (row.getAs[Int]("productId"), row.getAs[SparseVector]("features").toArray)
      }
    ).rdd.map{
      case (productId,features) => (productId,new DoubleMatrix(features))
    }

    //商品做笛卡尔积，两两配对，计算余弦相似度
    val productRecsDF = productFeatures.cartesian(productFeatures)
      //过滤掉商品自身
      .filter {
      case (productA, productB) => productA._1 != productB._1
    }
      //计算余弦相似度
      .map {
      case (productA, productB) => {
        val simScore = consinSim(productA._2, productB._2)
        (productA._1, (productB._1, simScore))
      }
    }
      //过滤相似分低于0.4的
      .filter(_._2._2 > 0.4)
      .groupByKey()
      .map {
        case (productId, resc) => {
          ProductRecs(productId, resc.toList.sortWith(_._2 > _._2)
            .map(x => Recommendation(x._1, x._2)))
        }
      }.toDF()
    //将商品相似度矩阵写入mongo
    storeData2Mongo(productRecsDF, CONTENT_PRODUCT_RECS_COLLECTION)

    sparkSession.close()
  }

  /**
    * 计算两个向量的余弦相似度
    *
    * @param matrix
    * @param matrix1
    * @return
    */
  def consinSim(matrix: DoubleMatrix, matrix1: DoubleMatrix): Double = {
    matrix.dot(matrix1) / (matrix.norm2() * matrix1.norm2())
  }

  /**
    *
    * @param df
    * @param collection_name
    * @param mongoConfig
    */
  def storeData2Mongo(df: DataFrame, collection_name: String)(implicit mongoConfig: MongoConfig): Unit = {
    df.write
      .option("uri", mongoConfig.mongoUri)
      .option("collection", collection_name)
      .format("com.mongodb.spark.sql")
      .mode("overwrite")
      .save()
  }

}

/**
  * product样例类
  * 3982
  * Fuhlen 富勒 M8眩光舞者时尚节能无线鼠标(草绿)
  * 1057,439,736B009EJN4T2
  * https://images-cn-4.ssl-images-amazon.com/images/I/31QPvUDNavL._SY300_QL70_.jpg
  * 外设产品|鼠标|电脑|办公
  * 富勒|鼠标|电子产品|好用|外观漂亮
  */
case class Product(productId: Int, name: String, imageUrl: String, categories: String, tags: String)

/**
  * 标准推荐对象样例类
  *
  * @param productId
  * @param score
  */
case class Recommendation(productId: Int, score: Double)

/**
  * 商品相似度列表样例类
  *
  * @param productId
  * @param recs
  */
case class ProductRecs(productId: Int, recs: Seq[Recommendation])


/**
  * mongodb配置样例类
  *
  * @param mongoUri mongo连接uri
  * @param mongodb  mongo连接数据库
  */
case class MongoConfig(mongoUri: String, mongodb: String)
